package com.simple.mqtt.server.handler;

import com.simple.mqtt.common.tool.LogTool;
import com.simple.mqtt.server.config.MqttBrokerConfig;
import com.simple.mqtt.server.domain.Message;
import com.simple.mqtt.server.processor.MqttBrokerProcessor;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateEvent;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;

/**
 * MQTT代理通道处理器
 * 处理传入的MQTT消息并管理客户端连接
 */
public class MqttBrokerHandler extends ChannelInboundHandlerAdapter {
    private final MqttBrokerConfig config;
    private final ExecutorService businessExecutor;
    private final ScheduledExecutorService scheduledExecutor;
    private final MqttBrokerProcessor processor;

    /**
     * 构造函数
     * @param config 代理配置
     * @param businessExecutor 业务线程池，用于消息处理
     * @param scheduledExecutor 调度执行器，用于异步任务
     */
    public MqttBrokerHandler(MqttBrokerConfig config, ExecutorService businessExecutor, ScheduledExecutorService scheduledExecutor) {
        this.config = config;
        this.businessExecutor = businessExecutor;
        this.scheduledExecutor = scheduledExecutor;
        this.processor = new MqttBrokerProcessor(config, scheduledExecutor);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof MqttMessage) {
            MqttMessage mqttMessage = (MqttMessage) msg;
            MqttMessageType messageType = mqttMessage.fixedHeader().messageType();

            // 使用业务线程池异步处理MQTT消息
            businessExecutor.execute(() -> {
                try {
                    processMessage(ctx, mqttMessage, messageType);
                } catch (Exception e) {
                    LogTool.error("Error processing MQTT message", e);
                    ctx.close();
                }
            });
        } else {
            ctx.fireChannelRead(msg);
        }
    }

    /**
     * 根据消息类型处理MQTT消息
     * @param ctx 通道上下文
     * @param mqttMessage MQTT消息
     * @param messageType 消息类型
     */
    private void processMessage(ChannelHandlerContext ctx, MqttMessage mqttMessage, MqttMessageType messageType) {
        switch (messageType) {
            case CONNECT:
                processor.processConnect(ctx, (MqttConnectMessage) mqttMessage);
                break;
            case PUBLISH:
                processor.processPublish(ctx, (MqttPublishMessage) mqttMessage);
                break;
            case PUBACK:
                processor.processPubAck(ctx, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
                break;
            case PUBREC:
                processor.processPubRec(ctx, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
                break;
            case PUBREL:
                processor.processPubRel(ctx, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
                break;
            case PUBCOMP:
                processor.processPubComp(ctx, (MqttMessageIdVariableHeader) mqttMessage.variableHeader());
                break;
            case SUBSCRIBE:
                processor.processSubscribe(ctx, (MqttSubscribeMessage) mqttMessage);
                break;
            case UNSUBSCRIBE:
                processor.processUnsubscribe(ctx, (MqttUnsubscribeMessage) mqttMessage);
                break;
            case PINGREQ:
                processor.processPingReq(ctx);
                break;
            case DISCONNECT:
                processor.processDisconnect(ctx);
                break;
            default:
                LogTool.warn("Unsupported message type: {}", messageType);
                break;
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            LogTool.debug("空闲状态事件，关闭连接");
            ctx.close();
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String clientId = ctx.channel().attr(MqttBrokerProcessor.CLIENT_ID_ATTR).get();
        if (clientId != null) {
            try {
                // 处理遗嘱消息（如果存在）
                Message willMessage = config.getMessageStore().getWillMessage(clientId);
                if (willMessage != null) {
                    processor.sendWillMessage(willMessage);
                    config.getMessageStore().clearWillMessage(clientId);
                }
            } catch (Exception e) {
                LogTool.error("Error sending will message for client {}", clientId, e);
            }

            // 清理客户端资源
            config.getChannelManager().removeChannel(clientId);
            config.getSessionManager().remove(clientId);
            LogTool.info("Client {} disconnected", clientId);
        }
        super.channelInactive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LogTool.error("Exception caught in MqttBrokerHandler", cause);
        ctx.close();
    }
}

